package com.youth.config.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName RMQConfig  https://www.jianshu.com/p/d21bafe3b9fd
 * https://blog.csdn.net/qq_35387940/article/details/100514134?depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromBaidu-1&utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromBaidu-1
 * @Description: TODO
 * @Author yjwu
 * @Date 2020/1/6 14:14
 **/
@Configuration
@Slf4j
public class RMQConfig {
    @Value("${spring.rabbitmq.addresses}")
    private String addresses;
    //    @Value("${spring.rabbitmq.port}")
//    private int port;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.template.mandatory}")
    private boolean mandatory;
    @Value("${spring.rabbitmq.publisher-confirms}")
    private Boolean publisherconfirms;
    @Value("${spring.rabbitmq.publisher-returns}")
    private Boolean publisherreturns;
    @Value("${spring.rabbitmq.listener.simple.concurrency}")
    private int concurrency;
    @Value("${spring.rabbitmq.listener.simple.max-concurrency}")
    private int maxConcurrency;
    @Value("${spring.rabbitmq.listener.simple.prefetch}")
    private int prefetch;
//    @Autowired
//    private RMQMessageListener orderToCommodityRMQListener;


    @Value("${seckillRMQExchangeName}")
    private String seckillRMQExchangeName;
    @Value("${seckillRMQQueueName}")
    private String seckillRMQQueueName;
    @Value("${seckillRMQRoutingKey}")
    private String seckillRMQRoutingKey;

    @Value("${logRMQExchangeName}")
    private String logRMQExchangeName;
    @Value("${logRMQQueueName}")
    private String logRMQQueueName;
    @Value("${logRMQRoutingKey}")
    private String logRMQRoutingKey;

    /**
     * @MethodName: topicRMQ
     * @Description: TODO 声明队列
     * @Param: * @param
     * @Return: {@link Queue}
     * @Author: yjwu
     * @Date: 2019/11/30 22:26
     **/
    @Bean(name = "sekillRMQQueue")
    public Queue sekillRMQQueue() {
        return new Queue(seckillRMQQueueName);
    }

    @Bean(name = "logRMQQueue")
    public Queue logRMQQueue() {
        return new Queue(logRMQQueueName);
    }

    /**
     * @MethodName: topicExchange
     * @Description: TODO 声明交换器
     * @Param: * @param
     * @Return: {@link TopicExchange}
     * @Author: yjwu
     * @Date: 2019/11/30 22:32
     **/
    @Bean(name = "seckillRMQExchange")
    public TopicExchange seckillRMQExchange() {
        return new TopicExchange(seckillRMQExchangeName, true, true);
    }

    @Bean(name = "logRMQExchange")
    public TopicExchange logRMQExchange() {
        return new TopicExchange(logRMQExchangeName, true, true);
    }

    /**
     * @MethodName: topicBinding
     * @Description: TODO 声明绑定关系
     * @Param: * @param
     * @Return: {@link Binding}
     * @Author: yjwu
     * @Date: 2019/11/30 22:33
     **/
    @Bean(name = "seckillRMQBinding")
    public Binding seckillRMQBinding() {
        return BindingBuilder.bind(sekillRMQQueue()).to(seckillRMQExchange()).with(seckillRMQRoutingKey);
    }

    @Bean(name = "logRMQBinding")
    public Binding logRMQBinding() {
        return BindingBuilder.bind(logRMQQueue()).to(logRMQExchange()).with(logRMQRoutingKey);
    }


    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(addresses);
//        cachingConnectionFactory.setPort(port);
        cachingConnectionFactory.setUsername(username);
        cachingConnectionFactory.setPassword(password);
//        如果需要进行消息回调，必须设置为true
        cachingConnectionFactory.setPublisherConfirms(publisherconfirms);
        cachingConnectionFactory.setPublisherReturns(publisherreturns);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
//        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
//    因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最有一次设置
//    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        rabbitTemplate.setMessageConverter(new RMQMessageConvert());
        rabbitTemplate.setMandatory(mandatory);

        /**
         * @MethodName: rabbitTemplate
         * @Description: TODO 消息返回确认
         * @Param:  * @param
         * @Return: {@link RabbitTemplate}
         * @Author: yjwu
         * @Date: 2020/4/9 11:20
         **/
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息主体: {}", message);
                log.info("回复编码: {}", replyCode);
                log.info("回复内容: {}", replyText);
                log.info("交换器: {}", exchange);
                log.info("路由键: {}", routingKey);
            }
        });

        /**
         * @MethodName: rabbitTemplate
         * @Description: TODO 消息确认监听
         * @Param:  * @param
         * @Return: {@link RabbitTemplate}
         * @Author: yjwu
         * @Date: 2020/4/9 11:21
         **/
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息唯一标识: {}", correlationData);
                log.info("确认状态: {}", ack);
                log.info("造成原因: {}", cause);
            }
        });
        return rabbitTemplate;
    }

    /**
     * @MethodName: simpleMessageListenerContainer
     * @Description: TODO 声明简单的消费者，配置消费者的手动确认
     * @Param: * @param connectionFactory
     * @Return: {@link SimpleMessageListenerContainer}
     * @Author: yjwu
     * @Date: 2019/11/30 22:36
     **/
//    @Bean
//    SimpleMessageListenerContainer simpleMessageListenerContainer( ConnectionFactory connectionFactory) {
//        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
//        simpleMessageListenerContainer.setConnectionFactory(connectionFactory);
//        //并发配置
//        simpleMessageListenerContainer.setConcurrentConsumers(concurrency);
//        simpleMessageListenerContainer.setPrefetchCount(prefetch);
//        simpleMessageListenerContainer.setMaxConcurrentConsumers(maxConcurrency);
//        //消息确认机制
//        Queue[] queues=new Queue[2];
//        queues[0]=sekillRMQQueue();
//        queues[1]=logRMQQueue();
//        simpleMessageListenerContainer.setQueues(queues);
//        //rmq默认是自动确认，这里配置为手动确认
//        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//
//        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageHandler());
//        // 指定消息转换器
//        //设置处理器的消费消息的默认方法,如果没有设置，那么默认的处理器中的默认方式是handleMessage方法
//        adapter.setDefaultListenerMethod("onMessage");
//        Map<String, String> queueOrTagToMethodName = new HashMap<>();
//        queueOrTagToMethodName.put("order", "onorder");
//        queueOrTagToMethodName.put("pay", "onpay");
//        queueOrTagToMethodName.put("order", "oninfo");
//        adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
//        simpleMessageListenerContainer.setMessageListener(adapter);
//
//        return simpleMessageListenerContainer;
//
//    }
}
